use crate::logical_plan::expr::{LogicalPlanExpr, Operator, LiteralValue};
use std::collections::VecDeque;
use crate::physical_plan::{PhysicalPlanTrait, logicalplan_expr_map_physicalplan_expr, logicalplan_func_map_physicalplan};
use crate::run_plan::{RunPlan, RunPlanPhysicalPlanFuncArg, RunPlanInstance};
use crate::conf_init::{MapResult, ResultErr};

pub mod work_node;
pub mod work_space;
mod series;
pub mod expr;
pub mod filter;
pub mod func;

/*
conf->ast->LogicalPlanExpr->RunPlan->PhysicalPlanTrait->PhysicalPlan

LogicalPlanExpr就是客户具体的表达式集合，大致分几类：
    1）filter判断类型
    2）简单hand处理类型
    3）复杂处理类型
physicalPlanTrait是physicalplan的抽象化，由具体的PhysicalPlan来执行具体动作
RunPlan 对PhysicalPlan的组合方式，决定了数据在不同PhysicalPlan之间的跳转方式
*/

pub fn logicalplan_map(expr: &LogicalPlanExpr)-> MapResult<Box<RunPlan>>{
    match expr{
        LogicalPlanExpr::Binary {left,op,right} => {
            let left = match logicalplan_expr_map_physicalplan_expr(left){
                MapResult::Ok(left) => {
                    RunPlanPhysicalPlanFuncArg::Normal(left)
                },
                MapResult::Err(_) => {
                    let t = logicalplan_map(left)?;
                    RunPlanPhysicalPlanFuncArg::OtherRunPlan(t)
                }
            };
            let right = match logicalplan_expr_map_physicalplan_expr(right){
                MapResult::Ok(right) => {
                    RunPlanPhysicalPlanFuncArg::Normal(right)
                },
                MapResult::Err(_) => {
                    let t = logicalplan_map(right)?;
                    RunPlanPhysicalPlanFuncArg::OtherRunPlan(t)
                }
            };

            let op_func = logicalplan_func_map_physicalplan(op.to_func_name())?;
            return MapResult::Ok(
                Box::new(RunPlan::Filter {left:Box::new(left),op:op_func,right:Box::new(right)}));
        },
        LogicalPlanExpr::Func {name, input} => {
            let mut func_args = VecDeque::new();
            for arg in input.iter(){
                match arg{
                    LogicalPlanExpr::Column(_) => {
                        let arg = logicalplan_expr_map_physicalplan_expr(arg)?;
                        func_args.push_back(RunPlanPhysicalPlanFuncArg::Normal(arg));
                    },
                    LogicalPlanExpr::Columns(t) => {
                        let arg = logicalplan_expr_map_physicalplan_expr(arg)?;
                        func_args.push_back(RunPlanPhysicalPlanFuncArg::Normal(arg));
                    },
                    LogicalPlanExpr::Literal(_) => {
                        let arg = logicalplan_expr_map_physicalplan_expr(arg)?;
                        func_args.push_back(RunPlanPhysicalPlanFuncArg::Normal(arg));
                    },
                    LogicalPlanExpr::LiteralConf(_) => {
                        let arg = logicalplan_expr_map_physicalplan_expr(arg)?;
                        func_args.push_back(RunPlanPhysicalPlanFuncArg::Normal(arg));
                    },
                    LogicalPlanExpr::Binary {left,op,right} => {
                        return MapResult::Err(ResultErr::new(format!("func arg cannot be Binary")));
                    },
                    LogicalPlanExpr::Func {name, input} => {
                        let arg_runplan = logicalplan_map(arg)?;
                        func_args.push_back(RunPlanPhysicalPlanFuncArg::OtherRunPlan(arg_runplan));
                    },
                    other => {
                        return MapResult::Err(ResultErr::new(format!("func arg not support expr: [{:?}]", other)));
                    },
                }
            }
            let func = logicalplan_func_map_physicalplan(name)?;
            return MapResult::Ok(Box::new(RunPlan::Func(func, func_args)));
        },
        LogicalPlanExpr::Not(t) => {
            let tmp = logicalplan_map(t)?;
            return MapResult::Ok(Box::new(RunPlan::Not(tmp)));
        },

        _ => {
            return MapResult::Err(
                ResultErr::new(
                    format!("LogicalPlan [{:?}] cannot map to RunPlan", expr)
                )
            );
        },
    }
}
